package com.gis.bigdata.spark02.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable



/**
 * @author LnnuUser
 * @create 2022-10-05-下午10:11
 */
object Scala_Test01_3 {

  /*
  日期，用户id，session ID，页面id，动作时间，搜索关键字，点击品类id，产品id，下单品类id、支付品类id，产品id，城市id
  2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19
  2019-07-17,95,26070e87-1ad7-49a3-8fb3-cc741facaddf,48,2019-07-17 00:00:10,null,16,98,null,null,null,null,19
   */
  def main(args: Array[String]): Unit = {
    // 需求1：Top10热门品类
    // TODO: TOP10
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc: SparkContext = new SparkContext(sparkConf)

    // TODO 1. 读取原始数据
    val actionRDD: RDD[String] = sc.textFile("/home/lnnu/IdeaProjects/spark/datas/user_visit_action.txt")
    // Q1. RDD重复使用
    // 2. 将数据转换结构

    val acc: HotCategoryAccumulator = new HotCategoryAccumulator
    sc.register(acc, "hotCategory")
    actionRDD.foreach(
      action => {
        val datas: Array[String] = action.split("_")
        if (datas(6) != "-1"){
          // 点击
          acc.add((datas(6), "click"))
        }else if (datas(8) != "null"){
          // 下单
          val ids: Array[String] = datas(8).split(",")
          ids.foreach(
            id => {
              acc.add((id, "order"))
            }
          )
        }else if (datas(10) != "null"){
          val ids: Array[String] = datas(10).split(",")
          ids.foreach(
            id => {
              acc.add((id, "pay"))
            }
          )
        }
      }
    )

    // 3. 将相同的品类id的数据进行分组聚合
    // （品类id， （点击，下单， 支付））
    val accValue: mutable.Map[String, HotCategory] = acc.value

    val categories: mutable.Iterable[HotCategory] = accValue.map(_._2)

    // 4. 将统计结果根据数量降序排列，取前10
    val sortRDD: List[HotCategory] = categories.toList.sortWith(
      (left, right) => {
        if (left.clickCnt > right.clickCnt) {
          true
        } else if (left.clickCnt == right.clickCnt) {
          if (left.orderCnt > right.orderCnt) {
            true
          } else if (left.orderCnt == right.orderCnt) {
            left.payCnt > right.payCnt
          }else{
            false
          }
        }else {
          false
        }
      }
    )

    // 5. 将结果采集到控制台打印

    sortRDD.take(10).foreach(println)

    sc.stop()
  }

  // 自定义累加器

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  /**
   * 1. 继承AccumulatorV2
   *    IN：输入， （品类id，行为类型）
   *    OUT：输出，mutable.Map[String, HotCategory]
   * 2. 重写方法
   */
  class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]{
    private val hcMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hcMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
      new HotCategoryAccumulator()
    }

    override def reset(): Unit = {
      hcMap.clear()
    }

    override def add(v: (String, String)): Unit = {
      val cid: String = v._1
      val actionType: String = v._2
      val category: HotCategory = hcMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click"){
        category.clickCnt += 1
      }else if (actionType == "order"){
        category.orderCnt += 1
      }else if (actionType == "pay"){
        category.payCnt += 1
      }

      hcMap.update(cid, category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map1: mutable.Map[String, HotCategory] = this.hcMap
      val map2: mutable.Map[String, HotCategory] = other.value

      map2.foreach{
        case (cid, hc) => {
          val category: HotCategory = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          map1.update(cid, category)
        }
      }
    }

    override def value: mutable.Map[String, HotCategory] = hcMap
  }
}
